// -*-mode:c++; coding:utf-8-*-
#ifndef _XBASE_SSTABLE_BLOCK_CACHE_HPP_
#define _XBASE_SSTABLE_BLOCK_CACHE_HPP_

#include <cassert>

#include <boost/noncopyable.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/pool/pool_alloc.hpp>

#include <atomic.h>

#include "sstable_block_cache_base.hpp"
#include "sstable_block_cache_manager.hpp"

namespace xbase
{

	struct adopt_sstable_block_cache_t {};
	static const adopt_sstable_block_cache_t adopt_sstable_block_cache = {};

	class sstable_block_cache
	{
		typedef boost::fast_pool_allocator<atomic_t> atomic_allocator_type;

	public:
		typedef cache_node cache_node_type;
		typedef cache_key cache_key_type;

		sstable_block_cache()
			: m_cache_node(NULL)
			, m_ref_count(NULL) {}

		sstable_block_cache(cache_node_type *node)
			: m_cache_node(NULL)
			, m_ref_count(NULL) {
			if(node != NULL)
			{
				m_cache_node = sstable_block_cache_manager::instance().checkout(node->m_key);
				if(m_cache_node != NULL)
				{
					m_ref_count = m_atomic_allocator.allocate();
					atomic_set(m_ref_count, 1);
				}
			}
		}

		sstable_block_cache(cache_node_type *node, adopt_sstable_block_cache_t)
			: m_cache_node(node)
			, m_ref_count(NULL) {
			if(m_cache_node != NULL)
			{
				assert(node->m_value.ref_count() > 0); // must be already checked-out
				m_ref_count = m_atomic_allocator.allocate();
				atomic_set(m_ref_count, 1);
			}
		}

		sstable_block_cache(cache_key_type::file_id_type file_id,
				    cache_key_type::block_index_type index)
			: m_cache_node(NULL)
			, m_ref_count(NULL) {
			const cache_key_type key(file_id, index);
			m_cache_node = sstable_block_cache_manager::instance().checkout(key);
			if(m_cache_node != NULL)
			{
				m_ref_count = m_atomic_allocator.allocate();
				atomic_set(m_ref_count, 1);
			}
		}

		sstable_block_cache(const sstable_block_cache &other)
			: m_cache_node(other.m_cache_node)
			, m_ref_count(other.m_ref_count)
			, m_atomic_allocator(other.m_atomic_allocator) {
			if(m_cache_node != NULL)
			{
				atomic_inc(m_ref_count);
			}
		}

		sstable_block_cache &operator=(const sstable_block_cache &other) {
			if(this == &other)
				return *this;
			release();
			m_cache_node = other.m_cache_node;
			m_ref_count = other.m_ref_count;
			if(m_cache_node != NULL)
			{
				atomic_inc(m_ref_count);
			}
			return *this;
		}

		~sstable_block_cache() {
			release();
		}

		bool operator!() const {
			return m_cache_node == NULL;
		}

		void release() {
			if(m_cache_node != NULL)
			{
				if(atomic_dec_and_test(m_ref_count))
				{
					const bool ok = sstable_block_cache_manager::instance().checkin(m_cache_node);
					assert(ok); (void)ok;
					m_cache_node = NULL;
					m_atomic_allocator.deallocate(m_ref_count);
					m_ref_count = NULL;
				}
			}
		}

		block_index_type &block_index() {
			return m_cache_node->m_value.m_block_index;
		}
		
		const block_index_type &block_index() const {
			return m_cache_node->m_value.m_block_index;
		}

		static inline
		bool is_cached(cache_key_type::file_id_type file_id,
			       cache_key_type::block_index_type index) {
			const cache_key_type key(file_id, index);
			return sstable_block_cache_manager::instance().is_cached(key);
		}

#ifndef NDEBUG
		std::size_t ref_count() const {
			return ((m_ref_count == NULL)
				? 0
				: atomic_read(m_ref_count));
		}

		cache_node_type *node() const {
			return m_cache_node;
		}
#endif	// NDEBUG

	private:
		cache_node_type *m_cache_node;
		atomic_t *m_ref_count;
		atomic_allocator_type m_atomic_allocator;
	};

	class sstable_block_buffer
		: private boost::noncopyable
	{
	public:
		typedef cache_node cache_node_type;
		typedef cache_key cache_key_type;

		sstable_block_buffer()
			: m_node(NULL) {}

		~sstable_block_buffer() {
			release();
		}

		void release() {
			if(m_node != NULL) { // not used buffer
				sstable_block_cache_manager::instance().deallocate(m_node);
				m_node = NULL;
			}
		}

		// 模拟basic_simple_buffer接口
		
		boost::asio::mutable_buffer prepare(std::size_t size) {
			release();
			m_node = sstable_block_cache_manager::instance().allocate(size);
			return m_node->m_value.data();
		}

		void commit(std::size_t size) {
			m_node->m_value.m_block_size += size;
			assert(m_node->m_value.m_block_size <= m_node->m_value.m_alloc_size);
		}

		std::size_t size() const {
			return (m_node == NULL) ? 0 : m_node->m_value.size();
		}

		// 真正的commit接口

		bool commit(cache_key_type::file_id_type file_id,
			    cache_key_type::block_index_type index) {
			if(m_node != NULL)
			{
				// fill node
				const cache_key_type key(file_id, index);
				m_node->m_key = key;
				const bool ok = sstable_block_cache_manager::instance().commit(m_node);
				if(ok)
				{
					m_node = NULL;
					return true;
				}
			}
			return false;
		}

		inline
		sstable_block_cache
		commit_and_checkout(cache_key_type::file_id_type file_id,
				    cache_key_type::block_index_type index) {
			if(m_node != NULL)
			{
				// fill node
				const cache_key_type key(file_id, index);
				m_node->m_key = key;
				const bool ok = sstable_block_cache_manager::instance().commit_and_checkout(m_node);
				if(ok)
				{
					sstable_block_cache block(m_node, adopt_sstable_block_cache);
					m_node = NULL;
					return block;
				}
			}
			return sstable_block_cache();
		}

		bool operator!() const {
			return m_node == NULL;
		}

		bool build_index() {
			return m_node->m_value.build_index();
		}

	private:
		cache_node_type *m_node;
	};

} // namespace xbase

#endif	// _XBASE_SSTABLE_BLOCK_CACHE_HPP_
